/*
 * Copyright (c) 2022-2023 cetcclout Co., Ltd All rights reserved.
 * Sensor Adapter is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package com.cetccloud.cimp.adapter.server.mqtt.subscribe;

import com.cetccloud.cimp.adapter.config.ApplicationProperties;
import com.cetccloud.cimp.adapter.server.mqtt.MqttSign;
import com.cetccloud.cimp.common.util.SensorStringUtils;
import lombok.RequiredArgsConstructor;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @author
 * @Description TODO 设置订阅服务配置变更连接
 * @Date 2022/7/12 9:49
 */
@Component
@RequiredArgsConstructor
public class MqttReloadServerConfigSubClient implements Runnable {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private final ApplicationProperties applicationProperties;
    private final MqttReloadServerConfigCallback mqttReloadServerConfigCallback;

    /**
     * 初始化MQTT链接客户端
     */
    public void subscribeServerStatus() {
        String broker = applicationProperties.getMqtt().getBroker();
        String clientId = SensorStringUtils.SUBSCRIBE_PREFIX + "ServerConfig-" + UUID.randomUUID();
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttAsyncClient asyncClient = new MqttAsyncClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setAutomaticReconnect(true);
            connOpts.setCleanSession(true);

            // 设置超时时间 单位为秒
            connOpts.setConnectionTimeout(5);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            connOpts.setKeepAliveInterval(30);

            connOpts.setUserName(MqttSign.getUsername());
            connOpts.setPassword(applicationProperties.getMqtt().getPwd().toCharArray());

            mqttReloadServerConfigCallback.setMqttAsyncClient(asyncClient);
            asyncClient.setCallback(mqttReloadServerConfigCallback);
            IMqttToken connToken = asyncClient.connect(connOpts);
            connToken.waitForCompletion(1000L * 20);
        } catch (MqttException e) {
            logger.warn("MqttException : {}", e.getMessage());
        }
    }

    /**
     * 执行初始化方法
     */
    @Override
    public void run() {
        this.subscribeServerStatus();
    }

}
